package com.edu.yx.app.App_03_DwdApp.Dwd_05_Exam;

import com.edu.yx.app.APP_01_BaseApp.BaseSqlApp;
import com.edu.yx.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static com.edu.yx.common.Constant.*;

public class Dwd_02_Exam_PaperQuestionDetail extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_02_Exam_PaperQuestionDetail().init(
                3052,
                2,
                "Dwd_02_Exam_PaperQuestionDetail"
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        //todo 由于需求只要求题目的答题次数，正确次数，正确率，以及独立答题的人数，用一个表即可test_exam_question

        //1.读取ods_db
        readOdsDb(tEnv,"Dwd_02_Exam_PaperQuestionDetail");
        Table paper_question_detail_table = tEnv.sqlQuery("select " +
                "cast(data['question_id'] as string) id," +
                "cast(data['user_id'] as string) user_id," +
                "data['is_correct'] is_correct," +
                "date_format(data['create_time'],'yyyy-MM-dd') date_id," +
                "`type`, " +
                "`old`," +
                " ts ," +
                " current_row_timestamp() as row_op_ts " +
                " from ods_db " +
                " where `database` = 'edu' " +
                " and `table` = 'test_exam_question' " +
                " and `type` = 'insert' "
        );
        tEnv.createTemporaryView("paper_question_detail_table",paper_question_detail_table);

        //2.创建连接kafka的表
        tEnv.executeSql("create table dwd_exam_paper_question_detail(" +
                "id string," +
                "user_id string," +
                "is_correct string," +
                "date_id string," +
                "`type` string, " +
                "`old` MAP<STRING, STRING>, " +
                "ts bigint," +
                "row_op_ts timestamp_ltz(3)" +
                ")" + SQLUtil.getKafkaSink(TOPIC_DWD_EXAM_PAPER_QUESTION_DETAIL)
        );

        paper_question_detail_table.executeInsert("dwd_exam_paper_question_detail");





    }
}
